package com.shujia.streaming

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.ReceiverInputDStream

import java.sql.{Connection, DriverManager, PreparedStatement}


object Demo8SaveToMysql {
  def main(args: Array[String]): Unit = {
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local[2]")
      .appName("提交命令执行")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    val sparkContext: SparkContext = sparkSession.sparkContext
    val streamingContext = new StreamingContext(sparkContext, Durations.seconds(5))

    val linesDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)

    linesDS.foreachRDD((rdd:RDD[String])=>{
      println("------------正在处理一批数据-------------------")
      println(s"该批次的分区数：${rdd.getNumPartitions}")

      /**
       * foreachPartition 每一个分区处理一次逻辑
       */
      rdd.foreachPartition((itr:Iterator[String])=>{
        println("------------数加 防伪码-------------------")
        //创建与数据库的连接对象
        //1、注册驱动
        Class.forName("com.mysql.jdbc.Driver")
        //2、获取数据库连接对象
        val conn: Connection = DriverManager.getConnection(
          "jdbc:mysql://master:3306/bigdata29?useUnicode=true&characterEncoding=utf-8&useSSL=false",
          "root",
          "123456"
        )
        //获取数据库操作对象
        val ps: PreparedStatement = conn.prepareStatement("insert into student values(?,?,?,?,?)")

        itr.foreach((line:String)=>{
          println("....正在处理一条数据....")

          //1500100019,娄曦之,24,男,理科三班
          val infos: Array[String] = line.split(",")
          val id: Int = infos(0).toInt
          val name: String = infos(1)
          val age: Int = infos(2).toInt
          val gender: String = infos(3)
          val clazz: String = infos(4)

          //给预编译对象传参
          ps.setInt(1,id)
          ps.setString(2,name)
          ps.setInt(3,age)
          ps.setString(4,gender)
          ps.setString(5,clazz)
          //执行sql语句
          ps.executeUpdate()
        })
        //释放资源
        ps.close()
        conn.close()
        println()
      })
    })

    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()

  }
}
